package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}

object Demo15MapPartitions {
  def main(args: Array[String]): Unit = {
    /**
     * mapPartitions：转换算子
     * 可以先对RDD的每个分区进行操作
     */

    val blacklist: List[String] = List[String]("1500100777", "1500100888", "1500100666")

    val conf: SparkConf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("Demo15MapPartitions")

    val sc: SparkContext = new SparkContext(conf)
    /**
     * Spark的代码是需要以分布式的环境进行运行的
     * 虽然代码都好像是写在main方法当中，但实际上应该可以分为两个部分：算子内部、算子外部
     * 算子内部是以Task的形式发送到Executor中去执行的
     * 算子外部是在Driver端执行的
     * 所以在算子内部如果使用了算子外部定义的变量，则该变量必须实现序列化
     * 而MySQL连接是不能被序列化的，所以就会引发问题 Task not serializable
     * Caused by: java.io.NotSerializableException: com.mysql.jdbc.JDBC42PreparedStatement
     *
     */

    val blacklistRDD: RDD[String] = sc.parallelize(blacklist)


    // 连接远程MySQL的数据库，其中有一张表student，通过学生id，获取详细的信息
    blacklistRDD
      .map(id => {
        /** *
         * 如果把MySQL相关的操作都放入Task，虽然可以解决Task没有被序列化的问题，但是又会导致性能问题
         * 因为来一条数据，就需要建立一次连接，频繁地创建销毁连接代价太大了
         * 就需要借助mapPartitions来优化性能
         */
        // 建立连接
        val conn: Connection = DriverManager.getConnection("jdbc:mysql://rm-bp1h7v927zia3t8iwho.mysql.rds.aliyuncs.com:3307/stu023", "shujia23", "123456")

        // 创建Statement
        val pSt: PreparedStatement = conn.prepareStatement("select id,name,age,gender,clazz from student where id = ?")

        var returnStr: String = ""
        // 设置参数
        pSt.setString(1, id)
        // 执行查询，获取返回的结果
        val rs: ResultSet = pSt.executeQuery()
        while (rs.next()) {
          val id: String = rs.getString("id")
          val name: String = rs.getString("name")
          val age: String = rs.getString("age")
          val gender: String = rs.getString("gender")
          val clazz: String = rs.getString("clazz")
          returnStr = s"$id,$name,$age,$gender,$clazz"
        }
        returnStr
      }).foreach(println)

    /**
     * 对每个RDD的分区进行处理
     * 一个分区一般对应一个切片，一个切片一般会有多条数据
     * 多条数据在Spark中一般会通过Iterator类型进行接收
     *
     * 由于Task是作用在每一个分区上的，mapPartitions算子内部的代码会被执行的次数，由分区数决定
     * 大大减少连接建立的次数，提高性能
     *
     * 适用于需要跟外部数据源建立连接，并且从外部数据源获取数据，然后有后续处理时就可以使用mapPartitions，减少连接建立的次数
     *
     * 类似的：foreachPartition
     * 适用于需要跟外部数据源建立连接，将数据写入外部数据源或者是并且从外部数据源获取数据，但没有后续处理时就可以使用foreachPartition，减少连接建立的次数
     */
    blacklistRDD.mapPartitions(iter => {
      // 建立连接
      val conn: Connection = DriverManager.getConnection("jdbc:mysql://rm-bp1h7v927zia3t8iwho.mysql.rds.aliyuncs.com:3307/stu023", "shujia23", "123456")

      // 创建Statement
      val pSt: PreparedStatement = conn.prepareStatement("select id,name,age,gender,clazz from student where id = ?")

      // 这里的map是由Scala集合提供的一个方法，并不是算子
      iter.map(id => {
        var returnStr: String = ""
        // 设置参数
        pSt.setString(1, id)
        // 执行查询，获取返回的结果
        val rs: ResultSet = pSt.executeQuery()
        while (rs.next()) {
          val id: String = rs.getString("id")
          val name: String = rs.getString("name")
          val age: String = rs.getString("age")
          val gender: String = rs.getString("gender")
          val clazz: String = rs.getString("clazz")
          returnStr = s"$id,$name,$age,$gender,$clazz"
        }
        returnStr
      })
    }).foreach(println)

  }

}
